package com.xxwu.messagequeue.mqserver;

import com.xxwu.messagequeue.common.*;
import com.xxwu.messagequeue.common.argumentsType.*;
import com.xxwu.messagequeue.mqserver.core.BasicProperties;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
public class BrokerServer {
    // 创建用于网络连接的 Socket
    private ServerSocket serverSocket;
    // 默认一个服务器只有一个虚拟主机
    private VirtualHost virtualHost = new VirtualHost("default");
    // 用于存储客户端与服务器建立的多次连接的 channel
    private HashMap<String, Socket> sessions = new HashMap<>();
    // 创建线程池用于客户端和服务器建立连接
    private ExecutorService workPool = Executors.newCachedThreadPool();
    // 记录变量，用于控制服务器运行和停止
    // 防止编译器进行优化，保证每次都是从内存中读取
    private volatile boolean running = true;

    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(port);
    }

    // 服务器开始运行
    public void start() throws IOException {
        log.info("BrokerServer start!");
        while(running){
            try {
                Socket clientSocket = serverSocket.accept();
                // 将建立连接的工作交给线程池
                workPool.submit(()->{
                    // 开始处理客户端的请求
                    processConnection(clientSocket);
                });
                }catch (SocketException e) {
                    log.error("服务器停止运行！");
            }
        }
    }

    // 关闭服务器，用于单元测试
    public void stop() throws IOException {
        running = false;
        workPool.shutdownNow();
        serverSocket.close();
//        channels.clear();
        log.info("BrokerServer stop!");
    }

    private void processConnection(Socket clientSocket){
        // 读取客户端发来的请求
        // 将响应写回客户端
        try(InputStream inputStream = clientSocket.getInputStream()){
            try(OutputStream outputStream = clientSocket.getOutputStream()){
                // 由于之前规定了自定义协议的报文格式，协议头携带了4个字节的type、四个字节的length以及length个字节的body
                // 所以这里需要使用 DataInputStream 和 DataOutputStream
                try(DataInputStream dataInputStream = new DataInputStream(inputStream)){
                    try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                        while(true){
                            // 读取请求
                            Request request = readRequest(dataInputStream);
                            // 处理请求
                            Responds responds = process(request, clientSocket);
                            // 将请求写回客户端
                            writeResponds(responds, dataOutputStream);
                            log.info("ServerResponds.type={},responds.payload={}", responds.getType(),responds.getPayload());
                            log.info("响应返回成功");
                        }
                    }
                }catch (EOFException | SocketException e){
                    log.info("运行正常结束！连接关闭");
                }
            }
        }catch(IOException | MqException | ClassNotFoundException e){
            log.error(e.getMessage());
            log.error("处理请求失败");
        }finally {
            // 处理完一次请求后要将这个 clientSocket 关闭
            try {
                clientSocket.close();
                // 对于一次连接可能创建多个 channel ,关闭连接后，要把所有的 channel 都删除
                clearCloseChannels(clientSocket);
            } catch (IOException e) {
                log.error(e.getMessage());
                log.error("关闭连接失败！");
            }
        }
    }



    private void writeResponds(Responds responds, DataOutputStream dataOutputStream) throws IOException {
        dataOutputStream.writeInt(responds.getType());
        dataOutputStream.writeInt(responds.getLength());
        dataOutputStream.write(responds.getPayload());
        // 切记要刷新缓冲区，保证数据已经成功发送了
        dataOutputStream.flush();
    }

    private Responds process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException {
        // 1 将请求中携带的 payload 进行反序列化
        // 并将请求类型转化成所有请求类型的父类进行解析
        // 下面根据具体的请求类型，将参数转化为对应的子类
        BasicArguments basicArguments = (BasicArguments) BinaryTool.frombytes(request.getPayload());
        // 2 根据前面规定好的协议 type 代表的含义对请求进行对应的处理
        int type = request.getType();
        // 设定 Responds 返回值结果
        boolean ok = true;
        if(type == 0x1){
            // 建立连接
            sessions.put(basicArguments.getChannelId(), clientSocket);
            log.info("建立连接成功！channelId={}",basicArguments.getChannelId());
        } else if(type == 0x2){
            // 关闭连接
            sessions.remove(basicArguments.getChannelId());
            log.info("连接关闭成功！channelId={}",basicArguments.getChannelId());
        } else if(type == 0x3){
            // 创建 exchange
            ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicArguments;
            ok = virtualHost.exchangeDeclare(arguments.getExchangeName(),arguments.getType(),arguments.getDurable(),
                            arguments.getAutoDelete(), arguments.getArgument());
            log.info("exchange 创建成功！exchangeName={}",arguments.getExchangeName());
        } else if(type == 0x4){
            // 销毁 exchange
            ExchangeDeleteArguments arguments = (ExchangeDeleteArguments) basicArguments;
            ok = virtualHost.exchangeDelete(arguments.getExchangeName());
            log.info("exchange 销毁成功！exchangeName={}",arguments.getExchangeName());
        } else if(type == 0x5){
            // 创建 queue
            QueueDeclareArguments arguments = (QueueDeclareArguments) basicArguments;
            ok = virtualHost.queueDeclare(arguments.getQueueName(), arguments.getDurable(), arguments.getAutoDelete()
                            , arguments.getExclusive(), arguments.getArgument());
            log.info("queue 创建成功！queueName={}",arguments.getQueueName());
        } else if(type == 0x6){
            // 销毁 queue
            QueueDeleteArguments arguments = (QueueDeleteArguments) basicArguments;
            ok = virtualHost.queueDelete(arguments.getQueueName());
            log.info("queue 销毁成功！queueName={}",arguments.getQueueName());
        } else if(type == 0x7){
            // 创建 binding
            QueueBindArguments arguments = (QueueBindArguments) basicArguments;
            ok = virtualHost.queueBind(arguments.getExchangeName(), arguments.getQueueName(), arguments.getBindingKey());
            log.info("binding 创建成功！exchangeName={}, queueName={}, bindingKey={}"
                            ,arguments.getExchangeName(),arguments.getQueueName(), arguments.getBindingKey());
        } else if(type == 0x8){
            // 销毁 binding
            QueueUnBindArguments arguments = (QueueUnBindArguments) basicArguments;
            ok = virtualHost.queueUnbind(arguments.getExchangeName(), arguments.getQueueName());
            log.info("binding 销毁成功 exchangeName={}, queueName={}",arguments.getExchangeName(),arguments.getQueueName());
        } else if(type == 0x9){
            // 发送 message
            BasicPublishArguments arguments = (BasicPublishArguments) basicArguments;
            ok = virtualHost.basicPublish(arguments.getExchangeName(), arguments.getRoutingKey()
                            , arguments.getBasicProperties(), arguments.getBody());
            log.info("message 发送成功");
        } else if(type == 0xa){
            // 订阅队列
            BasicConsumerArguments arguments = (BasicConsumerArguments) basicArguments;
            ok = virtualHost.basicConsumer(arguments.getConsumerTag(), arguments.getQueueName(),arguments.isAutoAck()
                            ,new Consumer(){
                        // 一旦有消费者订阅队列成功，那么当队列有消息时，就开始执行这里的回调函数
                        // 而回调函数的内容里包含着发送响应的功能，所以当回调函数被执行一次，响应就被发送一次
                        @Override
                        public void handleDelivery(String consumerTag, BasicProperties properties, byte[] body) throws IOException, MqException {
                            // 这里要执行的逻辑就是把消息原封不动的返回给消费者
                            // 1 要知道这里的 message 要发送给哪个客户端
                            // 这里的 consumerTag 实际上是 channelId，根据这个 channelId 去sessions中 找到对应的客户端

                            // 这里的消息发送方不一定是当前订阅队列的客户端，因为这个回调函数并不是在这里执行的，而是在消费消息的线程中进行的
                            // 所以这里要先通过 consumerTag 找到对应的客户端，才能发送消息

                            Socket clientSocket = sessions.get(consumerTag);
                            // 判断当前客户端是否有效
                            log.info("clientSocket={}",clientSocket);
                            if(clientSocket == null || clientSocket.isClosed()){
                                throw new MqException("客户端不存在或已关闭！");
                            }
                            // 2 构造一个消息用于返回
                            SubScribeReturns subScribeReturns = new SubScribeReturns();
                            subScribeReturns.setBody(body);
                            subScribeReturns.setConsumerTag(consumerTag);
                            subScribeReturns.setProperties(properties);
                            subScribeReturns.setOk(true);
                            // 这里的 channelId 就是 consumerTag
                            subScribeReturns.setChannelId(consumerTag);
                            // 由于这里返回的只是一个单独地响应，即一有消息就把这个消息推送给订阅者
                            // 所以不需要与请求对应上
                            subScribeReturns.setRid("");
                            // 将这个返回值序列化
                            byte[] payload = BinaryTool.tobytes(subScribeReturns);
                            Responds responds = new Responds();
                            // 服务器给客户端推送的消息，响应独有的
                            responds.setType(0xc);
                            responds.setLength(payload.length);
                            responds.setPayload(payload);
                            // 将返回值写回客户端, 这里的 dataOutputStream 不能关闭，因为到持续不断地往客户端发送消息
                            // 如果将 dataOutputStream 关闭，那么与之关联的 clientSocket也会关闭
                            DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
                            writeResponds(responds, dataOutputStream);
                        }
                    });
        } else if(type == 0xb){
            // 返回 ack
            BasicAckArguments arguments = (BasicAckArguments) basicArguments;
            ok = virtualHost.basicAck(arguments.getQueueName(), arguments.getMessageId());
            log.info("ack 返回成功！");
        }
        // 构造响应体
        BasicReturns basicReturns = new BasicReturns();
        basicReturns.setOk(ok);
        basicReturns.setChannelId(basicArguments.getChannelId());
        // 与 request 保持相同的 rid，这样就能够判断是不是对应的请求和响应
        basicReturns.setRid(basicArguments.getRid());
        Responds responds = new Responds();
        // 将响应体序列化
        byte[] payload = BinaryTool.tobytes(basicReturns);
        // 与上面得到的 type 对应
        responds.setType(type);
        responds.setLength(payload.length);
        responds.setPayload(payload);
        return responds;
    }

    private Request readRequest(DataInputStream dataInputStream) throws IOException, MqException {
        Request request = new Request();
        int type = dataInputStream.readInt();
        int length = dataInputStream.readInt();
        byte[] payload = new byte[length];
        int actual = dataInputStream.read(payload);
        if(actual != length){
            throw new MqException("当前读取到的 payload 有误");
        }
        request.setType(type);
        request.setLength(length);
        request.setPayload(payload);
        return request;
    }

    private void clearCloseChannels(Socket clientSocket) {
        // 用于记录当前要删除的 channelId
        List<String> toDelete = new ArrayList<>();
        for(Map.Entry<String, Socket> entry : sessions.entrySet()){
            if(entry.getValue() == clientSocket){
//                注意，在集合类中一边遍历一边删除是大忌，删除会破坏原本集合的结构，在删除后继续遍历很有可能造成空指针
//                sessions.remove(entry.getKey());
                toDelete.add(entry.getKey());
            }
        }
        for(String todelete : toDelete){
            sessions.remove(todelete);
        }
    }
}
